package com.clougence.cloudcanal.dataprocess.widetable;

import com.clougence.cloudcanal.sdk.api.CloudCanalProcessorV2;
import com.clougence.cloudcanal.sdk.api.JavaDsType;
import com.clougence.cloudcanal.sdk.api.ProcessorContext;
import com.clougence.cloudcanal.sdk.api.modelv2.*;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.Types;
import java.util.*;

/**
 * <a href="/wide-table/src/main/resource/sql/MySQLToMySQLWithDimUPdate_one_fact_two_dim.sql"> example sql</a>
 */
public class MySqlToMySqlWithDimUpdate_one_fact_two_dim implements CloudCanalProcessorV2 {

    Logger log = LoggerFactory.getLogger(MySqlToMySqlWithDimUpdate_one_fact_two_dim.class);

    private DataSource          srcDataSource;

    // 事实表定义(fact table definition)
    private SchemaInfo          factTable             = new SchemaInfo(null, "wide_table", "task_schedule");

    // 第一个维表定义(dimension table definition)
    private SchemaInfo          dimensionTable1st = new SchemaInfo(null, "wide_table", "worker");

    // 第二个维表定义(dimension table definition)
    private SchemaInfo          dimensionTable2nd        = new SchemaInfo(null, "wide_table", "console");

    // 第一个维表关联字段在事实表上的定义(1st dimension table's join key definition in fact table)
    private CustomFieldV2       factTableJoinKey1st = CustomFieldV2.buildField("worker_id", null, Types.INTEGER, false, false, true);

    // 第二个维表关联字段在事实表上的定义(2nd dimension table's join key definition in fact table)
    private CustomFieldV2       factTableJoinKey2nd      = CustomFieldV2.buildField("console_id", null, Types.INTEGER, false, false, true);

    // 第一个维表关联字段在维表上的字段定义(1st dimension table's join key definition in dimension table)
    private CustomFieldV2 dimensionTableJoinKey1st = CustomFieldV2.buildField("id", null, Types.INTEGER, true, false, true);

    // 第一个维表关联字段在维表上的字段定义(1st dimension table's join key definition in dimension table)
    private CustomFieldV2 dimensionTableJoinKey2nd = CustomFieldV2.buildField("id", null, Types.INTEGER, true, false, true);

    // 宽表上的字段映射到第一个维度表
    private Map<CustomFieldV2, CustomFieldV2> colMapping1st = new HashMap<>();

    // 眼表上的字段映射到第二个维度表
    private Map<CustomFieldV2, CustomFieldV2> colMapping2nd = new HashMap<>();

    @Override
    public void start(ProcessorContext context) {
        if (context.getSrcDsType().getContextDsType() != JavaDsType.JdbcDataSource) {
            throw new IllegalArgumentException("src ds type is not JdbcDataSource");
        }

        srcDataSource = (DataSource) context.getSrcRdbDs();

        colMapping1st.put(
                CustomFieldV2.buildField("worker_state", null, Types.VARCHAR, false, true, true),
                CustomFieldV2.buildField("state", null, Types.VARCHAR, false, true, true)
        );
        colMapping1st.put(
                CustomFieldV2.buildField("worker_name", null, Types.VARCHAR, false, true, true),
                CustomFieldV2.buildField("name", null, Types.VARCHAR, false, true, true)
        );

        colMapping2nd.put(
                CustomFieldV2.buildField("console_domain", null, Types.VARCHAR, false, true, true),
                CustomFieldV2.buildField("domain", null, Types.VARCHAR, false, true, true)
        );
        colMapping2nd.put(
                CustomFieldV2.buildField("console_state", null, Types.VARCHAR, false, true, true),
                CustomFieldV2.buildField("state", null, Types.VARCHAR, false, true, true)
        );

    }

    @Override
    public List<CustomData> process(CustomData data) {
        List<CustomData> re = new ArrayList<>();
        if (data.getSchemaInfo().equals(factTable)) {
            handleFactTable(re, data);
        } else if (data.getSchemaInfo().equals(dimensionTable1st)) {
            handleDimensionTable(colMapping1st, dimensionTableJoinKey1st, factTableJoinKey1st, re, data);
        } else if (data.getSchemaInfo().equals(dimensionTable2nd)) {
            handleDimensionTable(colMapping2nd, dimensionTableJoinKey2nd, factTableJoinKey2nd, re, data);
        } else {
            re.add(data);
        }

        return re;
    }

    protected void handleDimensionTable(Map<CustomFieldV2, CustomFieldV2> mapping, CustomFieldV2 dimensionTableJoinKey, CustomFieldV2 factTableJoinKey, List<CustomData> re, CustomData data) {
        re.add(data);

        if (data.getEventType() == EventTypeInSdk.UPDATE) {
            // 更新主表所有包括了维表 JoinKey 的字段
            List<CustomRecordV2> records = new ArrayList<>();

            for (CustomRecordV2 recordV2 : data.getRecords()) {
                LinkedHashMap<String, CustomFieldV2> factJoinKeys = new LinkedHashMap<>();
                LinkedHashMap<String, CustomFieldV2> factAddCols = new LinkedHashMap<>();

                CustomFieldV2 jf = recordV2.getBeforeColumnMap().get(dimensionTableJoinKey.getFieldName());
                CustomFieldV2 factJoinKey = CustomFieldV2
                    .buildField(factTableJoinKey.getFieldName(), jf.getValue(), factTableJoinKey.getSqlType(), true, jf.getValue() == null, false);

                factAddCols.put(factJoinKey.getFieldName(), factJoinKey);
                factJoinKeys.put(factJoinKey.getFieldName(), factJoinKey);
                for (CustomFieldV2 factCol : mapping.keySet()) {
                    CustomFieldV2 dimensionCol = mapping.get(factCol);
                    CustomFieldV2 af = recordV2.getAfterColumnMap().get(dimensionCol.getFieldName());
                    CustomFieldV2 newField = CustomFieldV2.buildField(factCol.getFieldName(), af.getValue(), factCol.getSqlType(), factCol.isKey(), af.getValue() == null, true);
                    factAddCols.put(newField.getFieldName(), newField);
                }

                CustomRecordV2 r = CustomRecordV2.buildRecord(factAddCols, factAddCols, factJoinKeys, factJoinKeys);
                records.add(r);
            }

            CustomData newD = new CustomData(factTable, EventTypeInSdk.UPDATE, records);
            re.add(newD);
        } else if (data.getEventType() == EventTypeInSdk.DELETE) {
            // 更新主表所有包括了维表 JoinKey 的字段为 null
            List<CustomRecordV2> records = new ArrayList<>();

            for (CustomRecordV2 recordV2 : data.getRecords()) {
                LinkedHashMap<String, CustomFieldV2> factJoinKeys = new LinkedHashMap<>();
                LinkedHashMap<String, CustomFieldV2> factAddCols = new LinkedHashMap<>();

                CustomFieldV2 jf = recordV2.getBeforeColumnMap().get(dimensionTableJoinKey.getFieldName());
                CustomFieldV2 factJoinKey = CustomFieldV2
                    .buildField(factTableJoinKey.getFieldName(), jf.getValue(), factTableJoinKey.getSqlType(), true, jf.getValue() == null, false);

                factAddCols.put(factJoinKey.getFieldName(), factJoinKey);
                factJoinKeys.put(factJoinKey.getFieldName(), factJoinKey);
                for (CustomFieldV2 factCol : mapping.keySet()) {
                    CustomFieldV2 newField = CustomFieldV2.buildField(factCol.getFieldName(), null, factCol.getSqlType(), factCol.isKey(), true, true);
                    factAddCols.put(newField.getFieldName(), newField);
                }

                CustomRecordV2 r = CustomRecordV2.buildRecord(factAddCols, factAddCols, factJoinKeys, factJoinKeys);
                records.add(r);
            }

            CustomData newD = new CustomData(factTable, EventTypeInSdk.UPDATE, records);
            re.add(newD);
        }

    }

    protected String genDimensionQuerySql(Map<CustomFieldV2, CustomFieldV2> mapping, SchemaInfo dimensionTable, CustomFieldV2 dimensionTableJoinKey) {
        StringBuilder sb = new StringBuilder("SELECT ");
        boolean first = true;

        for (CustomFieldV2 addCol : mapping.values()) {
            if (first) {
                first = false;
            } else {
                sb.append(",");
            }

            sb.append("`").append(addCol.getFieldName()).append("`");
        }

        sb.append(" FROM ");
        sb.append("`").append(dimensionTable.getSchema()).append("`");
        sb.append(".").append("`").append(dimensionTable.getTable()).append("`");
        sb.append(" WHERE ");
        sb.append("`").append(dimensionTableJoinKey.getFieldName()).append("`=?");
        return sb.toString();
    }

    protected void handleFactTable(List<CustomData> re, CustomData data) {
        String sql1st = genDimensionQuerySql(colMapping1st, dimensionTable1st, dimensionTableJoinKey1st);
        log.info("SQL for dimension 1st: {}", sql1st);
        String sql2nd = genDimensionQuerySql(colMapping2nd, dimensionTable2nd, dimensionTableJoinKey2nd);
        log.info("SQL for dimension 2nd: {}", sql2nd);
        for (CustomRecordV2 recordV2 : data.getRecords()) {
            CustomFieldV2 f1;
            CustomFieldV2 f2;
            switch (data.getEventType()) {
                case INSERT:
                case UPDATE:
                    f1 = recordV2.getAfterColumnMap().get(factTableJoinKey1st.getFieldName());
                    f2 = recordV2.getAfterColumnMap().get(factTableJoinKey2nd.getFieldName());
                    break;
                case DELETE:
                    f1 = recordV2.getBeforeColumnMap().get(factTableJoinKey1st.getFieldName());
                    f2 = recordV2.getBeforeColumnMap().get(factTableJoinKey2nd.getFieldName());
                    break;
                default:
                    throw new IllegalArgumentException("unsupported event type:" + data.getEventType());
            }

            execSql(colMapping1st, recordV2, f1, dimensionTableJoinKey1st, sql1st);
            execSql(colMapping2nd, recordV2, f2, dimensionTableJoinKey2nd, sql2nd);
        }

        re.add(data);
    }

    protected void execSql(Map<CustomFieldV2, CustomFieldV2> mapping, CustomRecordV2 recordV2, CustomFieldV2 fieldInstance, CustomFieldV2 dimensionTableJoinKey, String sql) {
        try (Connection conn = srcDataSource.getConnection(); PreparedStatement ps = conn.prepareStatement(sql)) {
            ps.setObject(1, fieldInstance.getValue(), dimensionTableJoinKey.getSqlType());
            try (ResultSet rs = ps.executeQuery()) {
                if (rs.next()) {
                    for (CustomFieldV2 factCol : mapping.keySet()) {
                        CustomFieldV2 dimensionCol = mapping.get(factCol);
                        String val = rs.getString(dimensionCol.getFieldName());
                        CustomFieldV2 cf = CustomFieldV2.buildField(factCol.getFieldName(), val, factCol.getSqlType(), factCol.isKey(), val == null, true);
                        recordV2.addField(cf);
                    }
                } else {
                    // add empty cols
                    for (CustomFieldV2 factCol :  mapping.keySet()) {
                        CustomFieldV2 cf = CustomFieldV2.buildField(factCol.getFieldName(), null, factCol.getSqlType(), factCol.isKey(), true, true);
                        recordV2.addField(cf);
                    }
                }
            }
        } catch (Exception e) {
            throw new RuntimeException("process error.msg:" + ExceptionUtils.getRootCauseMessage(e), e);
        }
    }

    @Override
    public void stop() {
        // do nothing
    }
}
